package com.wngbms;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;

import java.util.Arrays;

@Slf4j
public class DataMonitor {

    private ZooKeeper zk;
    private String znode;
    boolean dead;
    private DataMonitorListener listener;
    private byte[] prevData;

    private StatCallback statCallback = new StatCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            boolean exists;
            switch (rc) {
                case Code.Ok:
                    exists = true;
                    break;
                case Code.NoNode:
                    exists = false;
                    break;
                case Code.SessionExpired:
                case Code.NoAuth:
                    dead = true;
                    listener.closing(rc);
                    return;
                default:
                    zk.exists(znode, true, this, null);
                    return;
            }

            byte[] b = null;
            if (exists) {
                try {
                    b = zk.getData(znode, false, null);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return;
                }
            }

            if ((b == null && b != prevData) || (b != null && !Arrays.equals(prevData, b))) {
                listener.exists(b);
                prevData = b;
            }
        }
    };

    public DataMonitor(ZooKeeper zk, String znode, DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.listener = listener;
        zk.exists(znode, true, statCallback, null);
    }

    public void handle(WatchedEvent event) {
        String path = event.getPath();

        if (event.getType() == Watcher.Event.EventType.None) {
            switch (event.getState()) {
                case SyncConnected:
                    break;
                case Expired:
                    dead = true;
                    listener.closing(Code.SessionExpired);
                    break;
            }
        } else {
            if (path != null && path.equals(znode)) {
                zk.exists(znode, true, statCallback, null);
            }
        }
    }

}
